package day11

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 第4章 DStream转换
 * DStream上的操作与RDD的类似，分为Transformations(转换)和OutputOperations(输出)两种，此外转换操作中
 * 还有一些比较特殊的算子。如：updateStateByKey()、transform()以及各种Window相关的算子
 */
object Spark_Streaming_4 {
}

/**
 * 4.1 无状态转化操作
 * 无状态转化操作就是把简单的RDD转化操作应用到每个批次上，也就是转换DStream中的每个RDD。部分无状态转化操作如下：
 *
 * ①map()。对DStream中的每个元素应用给定函数，返回由各元素输出的元素组成的DStream
 *
 * ②flatMap()。对DStream中的每个元素应用给定函数，返回由各元素输出的迭代器组成的DStream
 *
 * ③filter()。返回由给定DStream中通过筛选的元素组成的DStream
 *
 * ④repartition()。改变DStream的分区数
 *
 * ⑤reduceByKey()。将每个批次中键相同的记录规约
 *
 * ⑥groupByKey()。将每个批次中的记录根据键分组。
 *
 * 需要记住的是，尽管这些函数看起来像是作用在整个流上一样，但事实上每个DStream在内部是由许多RDD(批次)组成，且无状态转换操作
 * 是分别应用到每个RDD上的
 *
 * 例如：reduceByKey()会规约每个时间区间中的数据，但不会规约不同区间之间的数据。在之前的wordcount程序中，我们只会统计几秒内接收到
 * 的数据的单词个数，而不会累加。
 *
 */
object Spark_Streaming_4_1 {

}

/**
 * 4.1.1 Transform
 *
 * Transform允许DStream上执行任意的RDD-to-RDD函数。即使这些函数并没有在DStream的API中暴露出来，通过该函数
 * 可以方便的扩展Spark API。该函数每一批次调度一次。其实也就是对DStream中的RDD应用转换
 */
object Spark_Streaming_4_1_1 {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf配置文件对象
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_4_1_1").setMaster("local[*]")
    // 创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 创建DStream
    val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", 9999)
    // 转换为RDD操作
    val wordCountDS: DStream[(String, Int)] = lineDStream.transform(
      rdd => {
        val value: RDD[(String, Int)] = rdd.flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKey(_ + _)
        value
      }
    )

    // 打印输出
    wordCountDS.print()

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 4.2 有状态转化操作
 *
 */
object Spark_Streaming_4_2 {

}

/*
 * 4.2.1 UpdateStateByKey
 *
 * updateStateByKey算子用于将历史结果应用到当前批次，该操作允许在使用新信息不断更新状态的同时能够保留他的状态。
 *
 * 有时，我们需要在DStream中跨批次维护状态(例如流计算中累加wordCount)。针对这种情况，updateStateByKey()为
 * 我们提供了对一个状态变量的访问，用于键值对形式的Stream。给定一个由(键, 事件)对构成的DStream，并传递一个指定如何
 * 根据新的事件更新每个键对应状态的函数，他可以构成出一个新的DStream，其内部数据为(键, 状态)对。
 *
 * updateStateByKey()的结果会是一个新的DStream，其内部的RDD系列是由每个时间区间对应的(键, 状态)对组成的。
 * 为使用这个功能，需要做下面的两步操作：
 * ①定义状态，状态可以是一个任意的数据类型
 * ②定义状态更新函数，用于函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
 * 使用updateStateByKey需要对检查点目录进行配置，会使用检查点来保存状态。
 *
 * 例如：每隔一段时间景点人流量变化(从程序启动开始，在原有递增)
 * 编写代码
 */
object Spark_Streaming_4_2_1 {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf对象
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_4_2").setMaster("local[*]")
    // 创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 设置检查点路径(用于保存状态)
    ssc.checkpoint(directory = "spark-api-base/checkpoint2")
    // 创建DStream
    val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", port = 9999)

    val value: DStream[(String, Int)] = lineDStream.flatMap(_.split(" "))
      .map((_, 1))
      .updateStateByKey((seq: Seq[Int], state: Option[Int]) => {
        val sum: Int = seq.sum + state.getOrElse(0)
        Option(sum)
      })

    value.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 4.2.2 Window Operations(窗口操作)
 *
 * Spark Streaming 也提供了窗口计算，允许执行转换操作作用在一个窗口内的数据。默认情况下，计算只对
 * 一个时间段内的RDD进行，有了窗口之后，可以把计算应用到一个指定的窗口内的所有RDD上，一个窗口可以包含
 * 多个时间段，基于窗口的操作会在一个比StreamingContext的批次间隔更长的时间范围内，通过整合多个批次的
 * 结果，计算出整个窗口的结果。所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。
 * 	窗口时长：计算内容的时间范围；
 * 	滑动时长：隔多久触发一次计算
 * 注意：这两者都必须为采集周期的整数倍。
 *
 * 例如：一小时人流量的变化，窗口（6秒）和间隔（3秒）不一致，不一定从程序启动开始
 * 需求：WordCount统计 3秒一个批次，窗口6秒，滑步3秒。
 *
 */
object Spark_Streaming_4_2_2 {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_4_2_2").setMaster("local[*]")
    // 创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 设置检查点
    ssc.checkpoint(directory = "spark-api-base/checkpoint2")
    // 创建DStream
    val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", port = 9999)
    // 扁平映射
    val flatMapDStream: DStream[String] = lineDStream.flatMap(_.split(" "))
    // 设置窗口大小，滑动的步长
    val windowDS: DStream[String] = flatMapDStream.window(windowDuration = Seconds(6), slideDuration = Seconds(3))
    // wordCount
    val reduceDS: DStream[(String, Int)] = windowDS.map((_, 1)).reduceByKey(_ + _)
    // 打印输出
    reduceDS.print()

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 4.2.3 关于Window的操作还有如下方法：
 *
 * 1）window(windowLength, slideInterval)
 * 基于对源DStream窗化的批次进行计算返回一个新的DStream
 *
 * 2）countByWindow(windowLength, slideInterval)
 * 返回一个滑动窗口计数流中的元素个数
 *
 * 3）countByValueAndWindow()
 * 返回的DStream则包含窗口中每个值的个数
 *
 * 4）reduceByWindow(func, windowLength, slideInterval)
 * 通过使用自定义函数整合滑动区间流元素来创建一个新的单元素流
 *
 * 5）reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
 * 当在一个(k,v)对的DStream上调用此函数，会返回一个新(k,v)对的DStream，此处通过对滑动窗口
 * 中批次数据使用reduce函数来整合每个key的value值
 *
 * 6）reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
 * 这个函数是上述函数的变化版本，每个窗口的reduce值都是通过用前一个窗的reduce值来递增计算。通过reduce
 * 进入到滑窗数据并“反向reduce”离开窗口的旧数据来实现这个操作。如果把3s的时间窗口当成一个池塘，池塘每一秒都会有鱼游进
 * 或者游出，那么第一个函数表示每游进来一条鱼，就在该类鱼的数量上累加。而第二个函数就是，每游出去一条鱼，就将该鱼的总数减去一。
 */
object Spark_Streaming_4_2_3 {

}